package cn._51doit.day04;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * 标准键盘布局
 * ┌───┐   ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┐
 * │Esc│   │ F1│ F2│ F3│ F4│ │ F5│ F6│ F7│ F8│ │ F9│F10│F11│F12│ │P/S│S L│P/B│  ┌┐    ┌┐    ┌┐
 * └───┘   └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┘  └┘    └┘    └┘
 * ┌───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───────┐ ┌───┬───┬───┐ ┌───┬───┬───┬───┐
 * │~ `│! 1│@ 2│# 3│$ 4│% 5│^ 6│& 7│* 8│( 9│) 0│_ -│+ =│ BacSp │ │Ins│Hom│PUp│ │N L│ / │ * │ - │
 * ├───┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─────┤ ├───┼───┼───┤ ├───┼───┼───┼───┤
 * │ Tab │ Q │ W │ E │ R │ T │ Y │ U │ I │ O │ P │{ [│} ]│ | \ │ │Del│End│PDn│ │ 7 │ 8 │ 9 │   │
 * ├─────┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴─────┤ └───┴───┴───┘ ├───┼───┼───┤ + │
 * │ Caps │ A │ S │ D │ F │ G │ H │ J │ K │ L │: ;│" '│ Enter  │               │ 4 │ 5 │ 6 │   │
 * ├──────┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴────────┤     ┌───┐     ├───┼───┼───┼───┤
 * │ Shift  │ Z │ X │ C │ V │ B │ N │ M │< ,│> .│? /│  Shift   │     │ ↑ │     │ 1 │ 2 │ 3 │   │
 * ├─────┬──┴─┬─┴──┬┴───┴───┴───┴───┴───┴──┬┴───┼───┴┬────┬────┤ ┌───┼───┼───┐ ├───┴───┼───┤ E││
 * │ Ctrl│    │Alt │         Space         │ Alt│    │    │Ctrl│ │ ← │ ↓ │ → │ │   0   │ . │←─┘│
 * └─────┴────┴────┴───────────────────────┴────┴────┴────┴────┘ └───┴───┴───┘ └───────┴───┴───┘
 *
 * @create: 2021-10-18 09:58
 * @author: 今晚打脑斧
 * @program: UnionDemo
 * @Description:
 *      uion: 将多个【类型一致】的数据流合并到一起，返回一个新的流
 *      connect : (将两个流连接成一个)没有条件
 *      iterate : 用来做数据流迭代计算（增强分布式for循环）
 *      project(java特有) : 投影,只针对于Tuple类的DataStream,可以实现project取出想要的部分字段(还可以给数据指定位置)
 **/
public class TransformationDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        projectStream(env);

//        iterationStream(env);
//        unionStream(env);
//        ConnectStream(env);

        env.execute();
    }

    /**
     * project(java特有)
     * project 投影,只针对于Tuple类的DataStream,可以实现project取出想要的部分字段(还可以给数据指定位置)
     * 投影:光照射物体,不同角度不同影子
     */
    private static void projectStream(StreamExecutionEnvironment env) {
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> provinceCityAndMoney  = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String line) throws Exception {
                String[] split = line.split(",");
                return Tuple3.of(split[0], split[1], Integer.parseInt(split[2]));
            }
        });
        provinceCityAndMoney.project(2,1).print();
    }

    /**
     * iterate : 用来做数据流迭代计算（增强分布式for循环）
     * 核心就是numbers.iterate()
     * 传入迭代的条件 : iteration.closeWith(feedback);
     */
    private static void iterationStream(StreamExecutionEnvironment env) {
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        //将数字转换成long类型
        SingleOutputStreamOperator<Long> numbers = lines.map(Long::parseLong);

        //调用iterate方法 DataStream -> IterativeStream
        //对numbers进行迭代（不停的输入int的数字）
        IterativeStream<Long> iteration = numbers.iterate();

        //IterativeStream -> DataStream
        //对迭代出来的数据进行运算 //对输入的数据应用更新模型，即输入数据的处理逻辑
        SingleOutputStreamOperator<Long> iterationBody = iteration.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                System.out.println("iterate input =>" + value);
                return value -= 3;
            }
        });
        //只要满足value > 0的条件，就会形成一个回路，重新的迭代，即将前面的输出作为输入，在进行一次应用更新模型，即输入数据的处理逻辑
        //继续迭代的条件
        //feedback 反馈
        SingleOutputStreamOperator<Long> feedback = iterationBody.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value > 0;
            }
        });
        /**
         * 对数据进行两次过滤
         * 一次是迭代逻辑运算
         * 一次是迭代的限制
         * 下面的closeWith就是将两个连接起来,相当于是以后增强for循环
         */
        //传入迭代的条件
        iteration.closeWith(feedback);

        //不满足迭代条件的最后要输出
        SingleOutputStreamOperator<Long> output = iterationBody.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value <= 0;
            }
        });
        //数据输出
        output.print("output value");
    }

    /**
     * Connect(将两个流连接成一个)没有条件
     * 最大的用处:可以让两个数据流共享状态
     */
    private static void ConnectStream(StreamExecutionEnvironment env) {
        DataStreamSource<String> lines1 = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Integer> num = lines1.map(Integer::parseInt);
        DataStreamSource<String> lines2 = env.socketTextStream("doit01", 9999);

        ConnectedStreams<Integer, String> connectedStreams = num.connect(lines2);
        SingleOutputStreamOperator<String> resStream = connectedStreams.map(new CoMapFunction<Integer, String, String>() {
            //定义共享状态（变量，存取数据，可以容错）

            //对第一个数量处理的方法
            @Override
            public String map1(Integer value) throws Exception {
                return value.toString();
            }

            //对第二个数量处理的方法
            @Override
            public String map2(String value) throws Exception {
                return value.toUpperCase();
            }
        });

        resStream.print();
    }

    /**
     *  uion: 将多个【类型一致】的数据流合并到一起，返回一个新的流
     */
    private static void unionStream(StreamExecutionEnvironment env) {
        DataStreamSource<String> lines1 = env.socketTextStream("doit01", 8888);
        DataStreamSource<String> lines2 = env.socketTextStream("doit01", 9999);

        DataStream<String> union = lines1.union(lines2);
        System.out.println(union.getParallelism());//获得的并行度为1,是socket的并行度,但是union后的新的union并行度也是与执行环境一致

//        SingleOutputStreamOperator<Integer> num1 = lines1.map(Integer::parseInt);
//        SingleOutputStreamOperator<Integer> num2 = lines2.map(Integer::parseInt);
//        DataStream<Integer> union = num1.union(num2);
//        System.out.println(union.getParallelism());//没指定并行度,就是与执行环境一致


        union.print();
    }
}
